package com.bupt.kgplatform.service.impl;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bupt.kgplatform.common.RetResult;
import com.bupt.kgplatform.common.TugraphUtil;
import com.bupt.kgplatform.component.DynamicDataSource;
import com.bupt.kgplatform.entity.KgDatabase;
import com.bupt.kgplatform.mapper.DatabaseDao;
import com.bupt.kgplatform.mapper.KgDatabaseMapper;
import com.bupt.kgplatform.service.KgDatabaseService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedCaseInsensitiveMap;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author zhuyunfei
 * @since 2023-05-16
 */
@Service
public class KgDatabaseServiceImpl extends ServiceImpl<KgDatabaseMapper, KgDatabase> implements KgDatabaseService {

    @Autowired
    KgDatabaseMapper kgDatabaseMapper;

    @Autowired
    DynamicDataSource dynamicDataSource;

    @Autowired
    DatabaseDao databaseDao;

    @Override
    public void newDB(String dbname, String database, String url, String port, String username, String password, String reviser) {
        // 在数据表记录信息
        KgDatabase db = new KgDatabase();
        db.setDbname(dbname);
        db.setDbnumber(UUID.randomUUID().toString());
        db.setCreateTime(LocalDateTime.now());
        db.setUpdateTime(LocalDateTime.now());
        db.setUrl(url);
        db.setPort(port);
        db.setUsername(username);
        db.setPassword(password);
        db.setReviser(reviser);
        db.setStatus("初始化");

        try {
            // 创建数据源
            DruidDataSource dataSource = new DruidDataSource();
            if (database.equals("MySQL")) {
                db.setType(0);
                dataSource.setUrl(String.format("jdbc:mysql://%s:%s/", url, port));
                dataSource.setUsername(username);
                dataSource.setPassword(password);
                dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");

                Map<Object, Object> dataSourceMap = dynamicDataSource.getDataSourceMap();
                dataSourceMap.put(db.getDbnumber(), dataSource);
                dynamicDataSource.setTargetDataSources(dataSourceMap);
                System.out.println("-- 创建MySQL datasource:" + db.getDbnumber() + " --");
            }
            kgDatabaseMapper.insert(db);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public boolean deleteDB(String dbnumber) {
        try {
            QueryWrapper<KgDatabase> wrapper = new QueryWrapper<>();
            wrapper.eq("dbnumber",dbnumber);
            kgDatabaseMapper.delete(wrapper);
            dynamicDataSource.getDataSourceMap().remove(dbnumber);
        }catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Override
    public void changeSource(String dbnumber){
        Map<Object, Object> dataSourceMap = dynamicDataSource.getDataSourceMap();
        DruidDataSource dataSource;

        if(dataSourceMap.containsKey(dbnumber)) { // 存在数据源，直接调用
            dataSource = (DruidDataSource) dataSourceMap.get(dbnumber);
            databaseDao.setJdbcTemplate(dataSource);
            System.out.println("--调用datasource:"+dbnumber+"--");
        } else { // 不存在，先调用后创建
            dataSource = new DruidDataSource();
            QueryWrapper<KgDatabase> wrapper = new QueryWrapper<>();
            wrapper.eq("dbnumber",dbnumber);
            KgDatabase bo = kgDatabaseMapper.selectOne(wrapper);
            String url = bo.getUrl();
            String port = bo.getPort();
            String username = bo.getUsername();
            String password = bo.getPassword();
            int type = bo.getType();
            if(type == 0) { // mysql
                dataSource.setUrl(String.format("jdbc:mysql://%s:%s/", url, port));
                dataSource.setUsername(username);
                dataSource.setPassword(password);
                dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
                dataSourceMap.put(dbnumber, dataSource);
                System.out.println("--创建并调用datasource:"+dbnumber+"--");
                databaseDao.setJdbcTemplate(dataSource);
            }
            dynamicDataSource.setTargetDataSources(dataSourceMap);
        }
    }

    @Override
    public List getdbs(String dbnumber) {
        QueryWrapper<KgDatabase> wrapper = new QueryWrapper<>();
        wrapper.eq("dbnumber",dbnumber);
        KgDatabase bo = kgDatabaseMapper.selectOne(wrapper);
        System.out.println("--------获取数据库："+bo.toString()+"--");
        int type = bo.getType();

        List<Map<String, Object>> res = new ArrayList<>();

        if(type == 0) { // mysql
            List<LinkedCaseInsensitiveMap> list = databaseDao.getdbs();
            for(int i=0; i<list.size(); i++) {
                Map<String, Object> map = new HashMap<>();
                String dbname = list.get(i).get("Database").toString();
                map.put("value", dbname);
                map.put("label", dbname);

                List<Map> tables = new ArrayList<>();
                try {
//                    databaseDao.selectdb(dbname);
                    List<Map<String, Object>> tempList = databaseDao.getTables(dbname);
                    String key = String.format("Tables_in_%s", dbname); //这是返回结果的key
                    for(int j=0; j<tempList.size(); j++) {
                        Map<String, Object> tempMap = tempList.get(j);
                        if(tempMap.containsKey(key)) {
                            Map<String, Object> table = new HashMap<>();
                            String tableName = tempMap.get(key).toString();
                            List<Map<String, Object>> tempListcol = databaseDao.getColumns(dbname,tableName);
                            List<Map> cols = new ArrayList<>();
                            for(int k=0; k<tempListcol.size(); k++) {
                                String propertyName = tempListcol.get(k).get("列名").toString();
                                Map<String, String> col = new HashMap<>();
                                col.put("value", propertyName);
                                col.put("label", propertyName);
                                cols.add(col);
                            }
                            table.put("value", tableName);
                            table.put("label", tableName);
                            table.put("children",cols);
                            tables.add(table);
                        }
                    }
                    map.put("children", tables);
                    res.add(map);
                }
                catch (Exception e) {
                    continue;
                }
            }
        }
        return res;
    }

    /**
     * mysql：自动化配置方式，录入实体schema
     * @param entity_tables 实体表list
     * @param db_number
     * @param graphname 数据源名 和Tugragh图名一样
     */
    @Override
    public boolean addVLSchema(String entity_tables, String db_number,String graphname) throws InterruptedException {
        JSONArray tables = JSON.parseArray(entity_tables);
        System.out.println("插入节点标签："+entity_tables+"---"+tables.size());
        List<Map<String, Object>> tempList;

        //每个表构建一个点，然后导入数据
        for(int i=0; i<tables.size(); i++) {
            long start = System.currentTimeMillis();
            JSONObject table = tables.getJSONObject(i);
            String dbname = table.getString("datasource");
            String tablename = table.getString("table");
            tempList = databaseDao.getColumns(dbname, tablename);
            List<Map<String, Object>> tempPri = databaseDao.getPrimary(dbname, tablename); // primarykey只有一个，里面应该只存在一个Map。其实是constraint_name=PRIMARY的列名才是主键。
            JSONArray propertys = new JSONArray();
            for(int j=0; j<tempList.size(); j++) {
                //解析构建节点属性
                JSONObject property = new JSONObject();
                String propertyName = tempList.get(j).get("列名").toString();
                String propertySQLType = tempList.get(j).get("类型").toString();
                //用于name属性显示
                if(propertyName.equals(table.getString("keyname"))){
                    JSONObject name = new JSONObject();
                    name.put("name","name");
                    name.put("type",typeTransformer(propertySQLType));
                    name.put("optional",false);
                    propertys.add(name);
                }
                property.put("name",propertyName);
                property.put("type",typeTransformer(propertySQLType));
                if(tempPri.get(0).get("列名").toString().equals(propertyName))
                    property.put("optional",false);
                else
                    property.put("optional",true);
                propertys.add(property);
            }
            System.out.println(propertys);
            // 创建vertexLabel的schema
            HttpStatus httpStatus = TugraphUtil.addNodeLabel(graphname, tablename, propertys, tempPri.get(0).get("列名").toString());
            if (!httpStatus.is2xxSuccessful())
                return false;
            long end = System.currentTimeMillis();
            System.out.println(tablename+" schema转换完成，耗时: "+ (end-start));

            d2gEntity(tablename,dbname,table.getString("keyname"),graphname);
        }
        QueryWrapper<KgDatabase> wrapper = new QueryWrapper<>();
        wrapper.eq("dbnumber",db_number);
        KgDatabase kgDatabase = kgDatabaseMapper.selectOne(wrapper);
        kgDatabase.setStatus("转换中");
        kgDatabaseMapper.updateById(kgDatabase);
        return true;
    }

    /**
     * mysql: 录入关系schema 外键的方式
     * @param entity_tables 关系表list
     * @param db_number
     * @param graphname
     */
    @Override
    public boolean addELSchema(String entity_tables, String db_number, String graphname) throws InterruptedException {
        JSONArray tables = JSON.parseArray(entity_tables);
        System.out.println("插入关系标签："+entity_tables+"---"+tables.size());
        List<Map<String, Object>> tempList;

        //每个表构建一个点，然后导入数据
        for(int i=0; i<tables.size(); i++) {
            long start = System.currentTimeMillis();
            JSONObject table = tables.getJSONObject(i);
            String dbname = table.getString("datasource");
            String tablename = table.getString("table");
            String[] head = new String[3];//存放头节点相关表结构信息 下标0关系表中外键列名 下标1关系表中外键对应的节点表表名 下标2关系表中外键对应的节点表列
            String[] tail = new String[3];//同上
            List<String[]> refs = getReference(dbname, tablename);
            for(String[] s : refs){
                if(s[0].equals(table.getString("head"))){
                    head = s;
                }
                else if(s[0].equals(table.getString("tail"))){
                    tail = s;
                }
            }
            tempList = databaseDao.getColumns(dbname, tablename);
            ArrayList<String> colnames = new ArrayList<>();
            JSONArray propertys = new JSONArray();
            for(int j=0; j<tempList.size(); j++) {
                //解析构建节点属性
                JSONObject property = new JSONObject();
                String propertyName = tempList.get(j).get("列名").toString();
                String propertySQLType = tempList.get(j).get("类型").toString();
                colnames.add(propertyName);
                if(propertyName.equals(head[0])||propertyName.equals(tail[0]))
                    continue;
                property.put("name",propertyName);
                property.put("type",typeTransformer(propertySQLType));
                property.put("optional",true);
                propertys.add(property);
            }
            // 创建edgeLabel的schema
            HttpStatus httpStatus = TugraphUtil.addEdgeLabel(graphname, tablename, propertys,head[1],tail[1]);
            if (!httpStatus.is2xxSuccessful())
                return false;
            long end = System.currentTimeMillis();
            System.out.println(tablename+" schema转换完成，耗时: "+ (end-start));

            d2gRelation(tablename,head,tail,colnames,dbname,graphname);
        }
        QueryWrapper<KgDatabase> wrapper = new QueryWrapper<>();
        wrapper.eq("dbnumber",db_number);
        KgDatabase kgDatabase = kgDatabaseMapper.selectOne(wrapper);
        kgDatabase.setStatus("转换中");
        kgDatabaseMapper.updateById(kgDatabase);
        return true;
    }



    /**
     * mysql：获取指定数据库、指定table的外键信息
     * @param dbname
     * @param tableNmae
     * @return
     */
    public List getReference(String dbname, String tableNmae) {
        List<Map<String, Object>> tempList = databaseDao.getReference(dbname, tableNmae);
        List<String[]> refs = new ArrayList<>();
        for(int i=0; i<tempList.size(); i++) {
            if(tempList.get(i).get("外键表名") == null) {
                continue;
            }
            String[] table_column = new String[3];
            String column = tempList.get(i).get("列名").toString();
            String ref_table = tempList.get(i).get("外键表名").toString();
            String ref_column = tempList.get(i).get("外键列名").toString();
            table_column[0] = column;
            table_column[1] = ref_table;
            table_column[2] = ref_column;
            refs.add(table_column);
        }
        return refs;
    }

    public String typeTransformer(String type) {

        if (type.startsWith("int")){
            return "int32";
        }
        if (type.equals("smallint")){
            return "int16";
        }
        if (type.equals("tinyint")){
            return "int8";
        }
        if (type.equals("bigint")){
            return "int64";
        }
        if (type.startsWith("varchar") || type.startsWith("enum")
                || type.equals("text") || type.equals("longtext") || type.equals("set") || type.startsWith("char")){
            return "string";
        }
        if (type.equals("date") || type.equals("year")){
            return "date";
        }
        if (type.equals("datetime") || type.equals("timestamp") ){
            return "datetime";
        }
        if (type.equals("float")){
            return "float";
        }
        if (type.equals("double") || type.startsWith("decimal")){
            return "double";
        }
        return "string";
    }

    /**
     * mysql: 实体数据转换
     * @param label 实体表
     * @param dbname 数据库名
     * @param keyname 图name列对应实际列名
     * @param graphname 图谱名
     */
    public void d2gEntity(String label, String dbname, String keyname,String graphname) {
        System.out.println("----d2g数据转换「" + label + "」标签----");
        new Thread(() -> {
            List<Map<String, Object>> tempCount = databaseDao.getCount(dbname, label);
            int count = new Long((long) tempCount.get(0).get("count(*)")).intValue();
            int times = (int) (Math.ceil(count/1000.0));
            final CountDownLatch countDownLatch = new CountDownLatch(times);
            int poolSize = Runtime.getRuntime().availableProcessors()*2;
            // 加入线程池，多线程处理数据转换
            ExecutorService threadPool = new ThreadPoolExecutor(poolSize, poolSize, 2L, TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(times), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
            long start_time = System.currentTimeMillis();
            for(int k=0; k<times; k++) {
                final int threadnum = k;
                threadPool.execute(() -> {
                    try {
                        // 批量导入处理
                        JSONArray values = new JSONArray(); // 批量导入
                        JSONArray fields = new JSONArray();
                        int start = threadnum*1000;
                        List<Map<String, Object>> tempList = databaseDao.getLimitData(dbname, label, start, 1000);
                        //Tugraph接口fields参数构建
                        Map<String, Object> onecol = tempList.get(0);
                        fields.addAll(onecol.keySet());
                        fields.add("name");

                        for(int j=0; j<tempList.size(); j++) {
                            Map<String, Object> tempProperties = tempList.get(j);
                            ArrayList<Object> onedata = new ArrayList<>();
                            onedata.addAll(tempProperties.values());
                            onedata.add(tempProperties.get(keyname));
                            values.add(onedata);
                        }
                        if(values.size()!=0&&fields.size()!=0) {
                            TugraphUtil.addNodesBatch(graphname,label,fields,values);
                        }
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            threadPool.shutdown();
            long end_time = System.currentTimeMillis();
            System.out.println(label + " 数据转换完成, 耗时" + (end_time-start_time));

        }, "转换主线程").start();
    }

    /**
     * mysql：关系数据转换
     * @param relation_table 关系表名
     * @param head 头节点表信息
     * @param tail 尾节点表信息
     * @param colnames 关系表列名列表
     * @param dbname 数据库名
     * @param graphname 图谱名
     */
    public void d2gRelation(String relation_table,String[] head,String[] tail,ArrayList<String> colnames, String dbname,String graphname) {
        System.out.println("----d2g数据转换关系标签「"+relation_table+"」----");

        new Thread(() -> {
            List<Map<String, Object>> tempCount = databaseDao.getCount(dbname, relation_table);
            int count = new Long((long) tempCount.get(0).get("count(*)")).intValue();
            int times = (int) (Math.ceil(count/1000.0));
            final CountDownLatch countDownLatch = new CountDownLatch(times);
            int poolSize = Runtime.getRuntime().availableProcessors()*2;
            // 加入线程池，多线程处理数据转换
            ExecutorService threadPool = new ThreadPoolExecutor(poolSize, poolSize, 2L, TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(times), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
            String column1 = head[0]; // 本表的列名
            String source_label = head[1]; // 关联的表名
            String source_column = head[2]; // 关联的表的列名
            String column2 = tail[0];
            String target_label = tail[1];
            String target_column = tail[2];

//            selectdb(dbname);
            List<Map<String, Object>> tempPri = databaseDao.getPrimary(dbname, source_label);
            String source_primary = String.valueOf(tempPri.get(0).get("列名"));
            tempPri = databaseDao.getPrimary(dbname, target_label);
            String target_primary = String.valueOf(tempPri.get(0).get("列名"));

            Map<Object, Object> source_cache = ref_exchange(databaseDao.getRefData(dbname, source_label, source_primary, source_column));
            Map<Object, Object> target_cache = ref_exchange(databaseDao.getRefData(dbname, target_label, target_primary, target_column));
            //导入配置
            for(int i=0;i<colnames.size();i++){
                if(colnames.get(i).equals(column1)){
                    colnames.set(i,"SRC_ID");
                }
                else if(colnames.get(i).equals(column2)) {
                    colnames.set(i, "DST_ID");
                }
            }
            JSONObject importdes = new JSONObject();
            JSONArray files = new JSONArray();
            JSONObject file = new JSONObject();
            file.put("header",0);
            file.put("format","CSV");
            file.put("label", relation_table);
            file.put("SRC_ID", source_label);
            file.put("DST_ID", target_label);
            file.put("columns", colnames);
            files.add(file);
            importdes.put("files",files);
            System.out.println(importdes);
            long start_time = System.currentTimeMillis();
            for(int k=0; k<times; k++) {
                final int threadnum = k;
                threadPool.execute(() -> {
                    try {
                        JSONArray edgeParams = new JSONArray(); // 批量导入
                        int st = threadnum*1000;
                        List<Map<String, Object>> tempList = databaseDao.getLimitData(dbname, relation_table, st,1000);
                        StringBuilder data = new StringBuilder();
                        for(int j=0; j<tempList.size(); j++) {
                            // 直接存在内存里
                            Object source_id = source_cache.get(tempList.get(j).get(column1));
                            Object target_id = target_cache.get(tempList.get(j).get(column2));

                            Map<String, Object> tempProperties = tempList.get(j);

                            for(String key: tempProperties.keySet()){
                                if(key.equals(column1)){
                                    data.append(source_id);
                                }
                                else if(key.equals(column2)){
                                    data.append(target_id);
                                }
                                else {
                                    data.append(tempProperties.get(key));
                                }
                                data.append(',');
                            }
                            data.deleteCharAt(data.length()-1);
                            data.append("\r\n");
                        }
                        if(data.length()!=0) {
                            System.out.println(data);
                            TugraphUtil.importText(graphname,importdes.toString(),data.toString(),',');
                        }
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            threadPool.shutdown();
            long end_time = System.currentTimeMillis();
            System.out.println(relation_table +" 数据转换完成, 耗时"+ (end_time-start_time));
        }, "转换主线程").start();
    }

    private Map<Object, Object> ref_exchange(List<Map<String, Object>> list) { // Map<Object, Object>
        Map<Object, Object> map = new HashMap<>();
        for (int i = 0; i < list.size(); i++) {
            map.put(list.get(i).get("外键列"), list.get(i).get("主键"));
//            redisUtil.set(String.valueOf(list.get(i).get("外键列")), list.get(i).get("主键"));
        }
        return map;
    }
}
